package com.niit.service

import com.niit.bean.Answer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream

/*
实时分析学生在线教育

 */
class EDUDataService {


  def dataAnalysis(answer:DStream[Answer]): Unit ={
    hotQuestionTop10(answer)
    gradeQuestionTop10(answer)
    hotQuestionWithSubjectTop10(answer)
    studentMinScoreWithQuestion(answer)
    subjectWithQuestionTop10(answer)
    studentMinScoreTop10Question(answer)
  }

  //需求一：统计Top10的热点题 统计出现次数最多的10的题目
  /*
     {"student_id":"学生ID_39",
      "textbook_id":"教材ID_1",
      "grade_id":"年级ID_4",
      "subject_id":"科目ID_2_语文",
      "chapter_id":"章节ID_chapter_3",
      "question_id":"题目ID_660",
      "score":4,
      "answer_time":"2024-05-27 09:51:38",
      "ts":"May 27, 2024 9:51:38 AM"}
   */
  private def hotQuestionTop10(answer:DStream[Answer]): Unit ={
    //得到所有题目 出现次数 然后所有题目中 根据总次数 找到 前10
     //1.(key:question_id,value:总次数)
     val mapDS: DStream[(String, Int)] = answer.map(line => {
       val question_id = line.question_id
       (question_id, 1)
     })
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)

    //2.在输出（foreachRDD）中排序(根据次数进行降序，取前10个)
    reduceDS.foreachRDD(rdd=>{ //   DStream[(String, Int)] ==> rdd[(String, Int)]  其实rdd 是一批数据
      val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(10)
      //如果后续要根据此数据进行离线分析，可以已将次数保存进数据库。这里我仅作演示，保存进数据库自行编写
      println("--------------需求一：统计Top10的热点题----------------")
      top10.foreach( println )
    })
  }

  //需求二：统计Top3的答题活跃年级
  /*
     {"student_id":"学生ID_39",
      "textbook_id":"教材ID_1",
      "grade_id":"年级ID_4",
      "subject_id":"科目ID_2_语文",
      "chapter_id":"章节ID_chapter_3",
      "question_id":"题目ID_660",
      "score":4,
      "answer_time":"2024-05-27 09:51:38",
      "ts":"May 27, 2024 9:51:38 AM"}

       6: 300
       3: 280
       4：268

   */
  private def gradeQuestionTop10(answer:DStream[Answer]): Unit ={
    //(grade_id,count)
    val mapDS: DStream[(String, Int)] = answer.map(line => {
      (line.grade_id, 1)
    })

    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey(_ + _)

    reduceDS.foreachRDD(rdd=>{
      val top10: Array[(String, Int)] = rdd.sortBy(_._2, false).take(3)
      println("--------------需求二：统计Top3的答题活跃年级----------------")
      top10.foreach( println )
    })
  }

  /*
   需求三：统计top10热点题并带上所属科目
   {"student_id":"学生ID_39",
      "textbook_id":"教材ID_1",
      "grade_id":"年级ID_4",
      "subject_id":"科目ID_2_语文",
      "chapter_id":"章节ID_chapter_3",
      "question_id":"题目ID_660",
      "score":4,
      "answer_time":"2024-05-27 09:51:38",
      "ts":"May 27, 2024 9:51:38 AM"
   */
  private def hotQuestionWithSubjectTop10(answer:DStream[Answer]): Unit ={
    //1. 确定 Key  question_id  subject_id    和 value  count
    /*
        question_ID         subject_ID          count
          1                     1(语文)           1
          1                     2(英语)           1
          2                      1                1
          3                      1                1
     */

    val mapDS: DStream[((String, String), Int)] = answer.map(line => {
      val question_id = line.question_id
      val subject_id = line.subject_id
      ((question_id, subject_id), 1) // ( (古诗词第一题 1 ,语文),1 )   ( (应用题第一题 1,数学),1 )
    })

    val reduceDS: DStream[((String, String), Int)] = mapDS.reduceByKey(_ + _)
    reduceDS.foreachRDD(rdd=>{
      val top5: Array[((String, String), Int)] = rdd.sortBy(_._2, false).take(10)
      println("--------------需求三：统计top10热点题并带上所属科目----------------")
      top5.foreach( println )
    })
  }
  /*
     需求四：统计top5学生最低得分的成绩，并带上题目
     {"student_id":"学生ID_39",
        "textbook_id":"教材ID_1",
        "grade_id":"年级ID_4",
        "subject_id":"科目ID_2_语文",
        "chapter_id":"章节ID_chapter_3",
        "question_id":"题目ID_660",
        "score":4,
        "answer_time":"2024-05-27 09:51:38",
        "ts":"May 27, 2024 9:51:38 AM"
     */
  private def studentMinScoreWithQuestion(answer:DStream[Answer]): Unit ={
    val mapDS: DStream[((String, String), Int)] = answer.map(line => {
      ((line.student_id, line.question_id), line.score)
    })

    mapDS.foreachRDD(rdd=>{
      val top5: Array[((String, String), Int)] = rdd.sortBy(_._2, true).take(5)
      println("--------------需求四：统计top5学生最低得分的成绩，并带上题目----------------")
      top5.foreach( println )
    })
  }
  /*
   需求五：实时统计每个科目的 Top10 热点题  groupByKey
   */
  private def subjectWithQuestionTop10(answer:DStream[Answer]): Unit ={
    //1. ((subject_id,question_id),count)
    val mapDS: DStream[((String, String), Int)] = answer.map(line => { // [ ((语文，古诗词第一题),1),((语文，古诗词第二题),1),((语文，古诗词第一题),1)]
      ((line.subject_id, line.question_id), 1)
    })
    // [ ((语文，古诗词第一题),2),((语文，古诗词第二题),3),((数学，应用题第一题),5)]
    val reduceDS: DStream[((String, String), Int)] = mapDS.reduceByKey(_ + _)

    //2. (subject_id,(question_id,count))
    // [ (语文，(古诗词第一题,2)),     (语文，(古诗词第二题,3)),               (数学，(应用题第一题,5)) ]
    val newMapDS: DStream[(String, (String, Int))] = reduceDS.map {
      case ((subject_id, question_id), count) => { //
        (subject_id, (question_id, count))
      }
    }
    //  [ key:语文 ,value:[(古诗词第一题,2),(古诗词第二题,3),……………………………… ] ]
    // [ key:数学 ,value:[(应用题第一题,5),(应用题第二题,2),(计算题第五题,20),…………………………………… ] ]
    val groupDS: DStream[(String, Iterable[(String, Int)])] = newMapDS.groupByKey()
    // map:整条数据（既有Key又有Value）进行遍历
    //mapValues: 只对 KV形式数据 的 value进行遍历
    groupDS.foreachRDD(rdd=>{
      val resData: RDD[(String, List[(String, Int)])] = rdd.mapValues(value => { // value ::[(古诗词第一题,2),(古诗词第二题,3),……………………………… ] ]
        val sort: List[(String, Int)] = value.toList.sortBy(_._2).reverse
        sort.take(10)

      })
      println("--------------需求五：实时统计每个科目的 Top10 热点题----------------")
      resData.foreach( println )
    })


  }

  /*
    需求六：统计每个学生得分最低的前10道题，并带上所属题目
    reduceByKey: 将相同的Key,进行Value相加
      ("北海",200) ("北海",300),("北海",400)  reduceByKey => ("北海",900)
    groupByKey:  将相同的Key,对值进行拼接
    ("北海",200) ("北海",300),("北海",400)  groupByKey =>("北海",[200,300,400])
   */
  private def studentMinScoreTop10Question(answer:DStream[Answer]): Unit ={
    //key :student_id             value question_id  score
    val mapDS: DStream[(String, (String, Int))] = answer.map(line => {
      (line.student_id, (line.question_id, line.score))
    })// [("张三",(古诗词第一题,5)),("张三",(古诗词第二题,2)),("张三",(古诗词第三题,8))  ]
    /*
    [("张三",(古诗词第一题,5)),("张三",(古诗词第二题,2)),("张三",(古诗词第三题,8))  ]
      groubyKey
      ("张三",[ (古诗词第一题,5), (古诗词第二题,2) ,(古诗词第三题,8) ])

      mapvalues
     */
    val groupDS: DStream[(String, Iterable[(String, Int)])] = mapDS.groupByKey()

    groupDS.foreachRDD(rdd=>{
      val resData: RDD[(String, List[(String, Int)])] = rdd.mapValues(value => {
        val top10: List[(String, Int)] = value.toList.sortBy(_._2).take(10)
        top10
      })

      println("--------------需求六：统计每个学生得分最低的前10道题，并带上所属题目----------------")
      resData.foreach( println )

    })



  }

}
